#pragma once
// c++libs
#include <iostream>
#include <algorithm>
#include <vector>
#include <functional>
#include <unordered_map>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
// clibs
#include <cstdio>
#include <cstring>
#include <ctime>
#include <cstdlib>
#include <cstdint>
#include <cerrno>
#include <cassert>
// syslibs
#include <unistd.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <signal.h>

using namespace std;


// log
#define INFO 1
#define WARNING 2
#define ERR 3
#define DEBUG 4
#define FATAL 5

#define LOG_LEVEL INFO

#define LOG(level, fmt, ...) do {\
    if(level < LOG_LEVEL) break;\
    time_t current_time = time(nullptr);\
    struct tm* time_info = localtime(&current_time);\
    char formatted_time[26];\
    strftime(formatted_time, sizeof(formatted_time), "%Y-%m-%d %H:%M:%S", time_info);\
    printf("[%d %s %s:%d] " fmt "\n", (int)pthread_self(), formatted_time, __FILE__, __LINE__, ##__VA_ARGS__);\
}while(0)

#define INFO_LOG(fmt, ...) LOG(INFO, fmt, ##__VA_ARGS__)
#define WARNING_LOG(fmt, ...) LOG(WARNING, fmt, ##__VA_ARGS__)
#define ERROR_LOG(fmt, ...)  LOG(ERR, fmt, ##__VA_ARGS__)
#define DEBUG_LOG(fmt, ...) LOG(DEBUG, fmt, ##__VA_ARGS__)
#define FATAL_LOG(fmt, ...) LOG(FATAL, fmt, ##__VA_ARGS__)


// buffer
#define DEFAULT_BUFFER_SIZE 1024
class buffer {
public:
    buffer(size_t size = DEFAULT_BUFFER_SIZE) 
        :_reader_idx(0), _writer_idx(0), _buffer(size)
    {}

    char *begin() {
        return &_buffer[0];
    }

    // 获取当前的写入位置
    char *get_write_pos() {
        return &_buffer[_writer_idx];
    }

    // 获取当前的读取位置
    const char *get_read_pos() const {
        return &_buffer[_reader_idx];
    }

    // 获取缓冲区末尾空闲空间大小
    uint32_t get_endofbuffer_size() const {
        return _buffer.size() - _writer_idx;
    }

    // 获取缓冲区首部空闲空间大小
    uint32_t get_beginofbuffer_size() const {
        return _reader_idx;
    }

    // 获取可读空间大小
    uint32_t get_readable_size() const {
        return _writer_idx - _reader_idx;
    }

    // 将读指针向后移动
    void move_read_pos(uint32_t len) {
        if(len == 0) {
            return;
        }
        if(len > get_readable_size()) {
            ERROR_LOG("Error: move_read_pos out of range");
            exit(-1);
        }
        _reader_idx += len;
    }

    // 将写指针向后移动
    void move_write_pos(uint32_t len) {
        if(len == 0) {
            return;
        }
        if(len > get_endofbuffer_size()) {
            cerr << "Error: move_write_pos out of range" << endl;
            exit(-1);
        }
        _writer_idx += len;
    }

    // 确保可写空间足够
    void make_enouth_space(uint32_t len) {
        // 如果剩余空间够什么也不做
        if(len <= get_endofbuffer_size()) {
            return;
        }

        // 如果整体空闲空间足够，则挪动数据
        if(len <= get_endofbuffer_size() + get_beginofbuffer_size()) {
            uint32_t len = get_readable_size();
            copy(get_read_pos(), get_read_pos() + len, begin());
            // 调整读写指针
            _reader_idx = 0;
            _writer_idx = len;
        }
        else {
            // 扩容，不要挪动数据，直接从写之后扩容即可
            size_t need_size = len + _writer_idx;
            size_t new_size = get_new_size(need_size);
            _buffer.resize(new_size);
        }
    }

    // 写入数据
    void writer(const void *buf, size_t len) {
        if(len == 0) {
            return;
        }
        make_enouth_space(len);
        const char *bf = reinterpret_cast<const char*>(buf);
        copy(bf, bf + len, get_write_pos());
        move_write_pos(len);
    }   

    void buffer_writer(const buffer &data) {
        writer(data.get_read_pos(), data.get_readable_size());
    }

    void string_writer(const string &s) {
        writer(s.c_str(), s.size());
    }

    // 读取数据
    void reader(void *buf, size_t len) {
        if(len > get_readable_size()) {
            cerr << "reader error" << endl;
            exit(-1);
        }
        char *bf = reinterpret_cast<char*>(buf);
        copy(get_read_pos(), get_read_pos() + len, bf);
        move_read_pos(len);
    }

    buffer buffer_reader(size_t len) {
        buffer data(len);
        reader(data.get_write_pos(), len);
        data.move_write_pos(len);
        return data;
    }

    string string_reader(size_t len) {
        string s;
        s.resize(len);
        reader(const_cast<char*>(s.c_str()), len);
        return s;
    }

    string find_line() {
        const char *line = find(get_read_pos(), const_cast<const char*>(get_write_pos()), '\n');
        if(line == get_write_pos()) {
            return "";
        }
        return string_reader(line - get_read_pos() + 1);
    }

    void clear() {
        _reader_idx = _writer_idx = 0;
    }

private:
    // 快速找出一个合适n，使得2^n满足大于等于new_size的最小值
    size_t get_new_size(size_t need_size) const {
        size_t new_size = 1;
        while(new_size < need_size) {
            new_size <<= 1;
        }
        return min(new_size, _buffer.max_size());
    }

private:
    vector<char> _buffer;
    // 注意这里的读写指针只是一个相对于这块空间起始位置的一个偏移量
    // 并不是真正的物理内存地址
    // 写入和获取数据时都需要物理地址
    uint32_t _reader_idx;
    uint32_t _writer_idx;
};


// socket
const string default_ip = "0.0.0.0";
const uint16_t default_port = 8080;

class Socket {
    static const int default_backlog = 5;
public:
    Socket() :_sockfd(-1) {}

    Socket(int fd) : _sockfd(fd) {}

    ~Socket() {
        close();
    }

    void socket() {
        _sockfd = ::socket(AF_INET, SOCK_STREAM, 0);
        if(_sockfd < 0) {
            FATAL_LOG("socket error: %s", strerror(errno));
            exit(errno);
        }
        set_sock_reuseaddr();
        INFO_LOG("socket success, _sockfd = %d", _sockfd);
        
    }

    void bind(uint16_t port = default_port) {
        struct sockaddr_in localhost;
        localhost.sin_family = AF_INET;
        localhost.sin_port = htons(port);
        localhost.sin_addr.s_addr = INADDR_ANY;
        if(::bind(_sockfd, (const struct sockaddr*)&localhost, sizeof(localhost)) < 0) {
            FATAL_LOG("bind error: %s", strerror(errno));
            exit(errno);
        }
        INFO_LOG("bind success");
    }

    void listen() {
        if(::listen(_sockfd, default_backlog) < 0) {
            FATAL_LOG("listen error: %s", strerror(errno));
            exit(errno);
        }
        INFO_LOG("listen success");
    }

    int accept(bool print_client_info = true) {
        struct sockaddr_in client_addr; 
        socklen_t client_len = sizeof(client_addr);
        int newfd = ::accept(_sockfd, (struct sockaddr *)&client_addr, &client_len);
        if(newfd < 0) {
            WARNING_LOG("accept error: %s", strerror(errno));
            return -1;
        }
        if(print_client_info) {
            char ip_buff[INET_ADDRSTRLEN];
            if(inet_ntop(client_addr.sin_family, &client_addr.sin_addr, ip_buff, sizeof(ip_buff)) != nullptr) {
                INFO_LOG("get a new link, client_ip = %s, clinet_port = %d", ip_buff, ntohs(client_addr.sin_port));
            }
        }
        INFO_LOG("accept success, newfd = %d", newfd);
        return newfd;
    }

    void connect(const string &ip, uint16_t port) {
        struct sockaddr_in serverhost;
        serverhost.sin_family = AF_INET;
        serverhost.sin_port = htons(port);
        
        if(inet_pton(AF_INET, ip.c_str(), &(serverhost.sin_addr)) != 1) {
            FATAL_LOG("inet_pton() failed");
            exit(errno);
        }
        if(::connect(_sockfd, (const struct sockaddr *)&serverhost, sizeof(serverhost)) < 0) {
            FATAL_LOG("connect error: %s", strerror(errno));
            exit(errno);
        }
        INFO_LOG("connect success");
    }

    void close() {
        if(_sockfd != -1) {
            ::close(_sockfd);
        }
    }

    int get_sockfd() const {
        return _sockfd;
    }

    ssize_t recv(void *buf, size_t len, bool blocked = true) const {
        if(!blocked) {
            set_nonblock(_sockfd);
        }
        ssize_t n = ::recv(_sockfd, buf, len, 0);
        if(n <= 0) {
            if(errno == EINTR || errno == EWOULDBLOCK) {
                return 0;
            }
            else if(n == 0) {
                WARNING_LOG("client closed");
            }
            else {
                ERROR_LOG("recv error: %s", strerror(errno));
            }
            return -1;
        }
        return n;
    }

    ssize_t send(const void *buf, size_t len, bool blocked = true) const {
        if(len == 0) {
            return 0;
        }
        if(!blocked) {
            set_nonblock(_sockfd);
        }
        ssize_t n = ::send(_sockfd, buf, len, 0);
        if(n <= 0) {
            if(errno == EINTR || errno == EWOULDBLOCK) {
                return 0;
            }
            else if(n == 0) {
                WARNING_LOG("client closed");
            } 
            else {
                ERROR_LOG("send error: %s", strerror(errno));
            }
            return -1;
        }
        return n;
    }

    void set_nonblock(int fd) const {
        int flag = fcntl(fd, F_GETFL);
        if(flag == -1) {
            ERROR_LOG("Failed to set flag");
            return;
        }
        if (fcntl(fd, F_SETFL, flag | O_NONBLOCK) == -1) {  
            ERROR_LOG("fcntl F_SETFL");  
            return;  
        }
    }

    void create_server(uint16_t port = default_port) {
        socket();
        set_sock_reuseaddr();
        bind(port);
        listen();
    }

    void create_client(const string &ip, uint16_t port) {
        socket();
        connect(ip, port);
    }

    void set_sock_reuseaddr() const {
        int yes = 1;
        if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) == -1) {  
            FATAL_LOG("setsockopt error: %s", strerror(errno));
            exit(errno);  
        }
    }

private:
    int _sockfd;
};


// channel
class epoller;
class event_loop;
class channel {
    using event_callback = std::function<void()>;
public:
    channel(event_loop *ev_loop, int fd)
        :_ev_loop(ev_loop), _fd(fd), _events(0), _revents(0) 
    {}

    ~channel() {
        if(_fd != -1) {
            INFO_LOG("close channel fd: %d", _fd);
            close(_fd);
            _fd = -1;
        }
    }

    int get_fd() const { 
        return _fd; 
    }
    
    //获取想要监控的事件
    uint32_t get_events() const { 
        return _events; 
    }

    //设置实际就绪的事件
    void set_revents(uint32_t events) { 
        _revents = events; 
    }
    
    void set_read_callback(const event_callback &cb) {
        _read_callback = cb; 
    }

    void set_write_callback(const event_callback &cb) { 
        _write_callback = cb; 
    }

    void set_error_callback(const event_callback &cb) { 
        _error_callback = cb; 
    }

    void set_close_callback(const event_callback &cb) { 
        _close_callback = cb; 
    }

    void set_event_callback(const event_callback &cb) { 
        _event_callback = cb; 
    }

    //当前是否监控了可读
    bool readable() const { 
        return (_events & EPOLLIN); 
    } 

    //当前是否监控了可写
    bool writeable() const { 
        return (_events & EPOLLOUT); 
    }

    //启动读事件监控
    void enable_read() { 
        _events |= EPOLLIN; 
        update(); 
    }

    //启动写事件监控
    void enable_write() { 
        _events |= EPOLLOUT; 
        update(); 
    }

    //关闭读事件监控
    void DisableRead() { 
        _events &= ~EPOLLIN; 
        update(); 
    }

    //关闭写事件监控
    void disable_write() { 
        _events &= ~EPOLLOUT; 
        update(); 
    }

    //关闭所有事件监控
    void disable_all() { 
        _events = 0; 
        update(); 
    }

    //移除监控
    void remove();

    void update();

    //事件处理，一旦连接触发了事件，就调用这个函数，自己触发了什么事件如何处理自己决定
    void handle_event() const {
        if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)) {
            /*不管任何事件，都调用的回调函数*/
            if (_event_callback) _event_callback();
            if (_read_callback) _read_callback();
        }
        /*有可能会释放连接的操作事件，一次只处理一个*/
        if (_revents & EPOLLOUT) {
            if (_event_callback) _event_callback();
            if (_write_callback) _write_callback();
        }
        else if (_revents & EPOLLERR) {
            if (_event_callback) _event_callback();
            //一旦出错，就会释放连接，因此要放到前边调用任意回调
            if (_error_callback) _error_callback();
        }
        else if (_revents & EPOLLHUP) {
            if (_event_callback) _event_callback();
            if (_close_callback) _close_callback();
        }
    }
    
private:
    event_loop *_ev_loop; 
    int _fd;
    uint32_t _events;  // 当前需要监控的事件
    uint32_t _revents; // 当前连接触发的事件

    event_callback _read_callback;   //可读事件被触发的回调函数
    event_callback _write_callback;  //可写事件被触发的回调函数
    event_callback _error_callback;  //错误事件被触发的回调函数
    event_callback _close_callback;  //连接断开事件被触发的回调函数
    event_callback _event_callback;  //任意事件被触发的回调函数
};


// poller
class epoller {
    static const size_t default_events_size = 1024;
public:
    epoller(size_t events_size = default_events_size)
        :_size(events_size)
    {
        _epfd = epoll_create(1);
        if (_epfd == -1) {
            ERROR_LOG("epoll_create err");
            exit(errno);
        }
    }

    ~epoller() {
        close(_epfd);
    }

    vector<channel *> poller(int timeout = -1) {
        int nfds = ::epoll_wait(_epfd, _evs, _size, timeout);
        if (nfds == -1) {
            ERROR_LOG("epoll_wait err, err string: %s", strerror(errno));
            exit(errno);
        }
        vector<channel *> actives(nfds);
        for(int i = 0; i < nfds; ++i) {
            if(_channels.count(_evs[i].data.fd) == 0) {
                ERROR_LOG("epoller::poller, channel not found");
                continue;
            }
            auto ch_ptr = _channels[_evs[i].data.fd];
            ch_ptr->set_revents(_evs[i].events);
            actives[i] = ch_ptr;
        }
        return actives;
    }

    void update(channel *ch) {
        bool in_channels = channel_in_channels(ch);
        if(!in_channels) {
            _update(EPOLL_CTL_ADD, ch);
            _channels[ch->get_fd()] = ch;
        }
        else {
            _update(EPOLL_CTL_MOD, ch);
        }
    }

    void remove(const channel *ch) {
        if(!channel_in_channels(ch)) {
            return;
        }
        _update(EPOLL_CTL_DEL, ch);
        _channels.erase(ch->get_fd());
    }
    
    int get_epfd() {
        return _epfd;
    }

private:
    bool channel_in_channels(const channel *ch) const {
        return _channels.count(ch->get_fd()) > 0;
    }

    void _update(int op, const channel *ch) const {
        struct epoll_event ev;
        ev.data.fd = ch->get_fd();
        ev.events = ch->get_events();
        if (epoll_ctl(_epfd, op, ch->get_fd(), &ev) == -1) {
            ERROR_LOG("epoll_ctl err, op: %d, fd: %d, err string: %s", op, ch->get_fd(), strerror(errno));
            exit(errno);
        }
    }

private:
    int _epfd;
    struct epoll_event _evs[default_events_size];
    size_t _size;
    unordered_map<int, channel*> _channels;
};

// 任务类
using destroy_func_t = function<void()>;
using release_func_t = destroy_func_t;
class timer_task {
public:
    timer_task(uint64_t id, uint32_t delay, const destroy_func_t &cb) 
        : _id(id), _timeout(delay), _cancelled(false), _destroy_cb_(cb) 
    {}

    uint32_t get_timeout() const {
        return _timeout;
    }

    uint64_t get_id() const {
        return _id;
    }

    void set_release_cb(const release_func_t &cb) {
        _release_cb = cb;
    }

    void cancel_task() {
        _cancelled = true;
    }

    bool is_canceled() const {
        return _cancelled;
    }

    ~timer_task() {
        // 只有被没被取消才销毁
        if(!_cancelled) {
            _destroy_cb_();
            _release_cb();
        }
        
    }
private:
    // 每一个定时器任务得有表示唯一标识id
    // 由于服务器是一直运行的，可能有很多的任务，所以uint64_t
    // 尽量由系统统一分配
    uint64_t _id;
    // 更新超时时间的基准值
    uint32_t _timeout;
    // 任务是否被取消
    bool _cancelled;
    // 定时器要执行的销毁任务
    destroy_func_t _destroy_cb_;
    release_func_t _release_cb;
};


class time_wheel {
    using task_sptr = shared_ptr<timer_task>;
    using task_wptr = weak_ptr<timer_task>;
private:
    static int get_timerfd() {
        int timer_fd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
        if(timer_fd == -1) {
            ERROR_LOG("timerfd_create err, err string: %s", strerror(errno));
            exit(errno);
        }
        //int timerfd_settime(int fd, int flags, struct itimerspec *new, struct itimerspec *old);
        struct itimerspec itime;
        itime.it_value.tv_sec = 1; //第一次超时时间为1s后
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1; //第一次超时后，每次超时的间隔时
        itime.it_interval.tv_nsec = 0; 
        if(timerfd_settime(timer_fd, 0, &itime, nullptr) == -1) {
            ERROR_LOG("timerfd_settime err, err string: %s", strerror(errno));
            exit(errno);
        }
        return timer_fd;
    }

    uint64_t read_timerfd() {
        uint64_t times = 0;
        ssize_t n = read(_timerfd, &times, sizeof(times));
        if(n != sizeof(times)) {
            ERROR_LOG("read timerfd err, err string: %s", strerror(errno));
            exit(errno);
        }
        return times;
    }

    void add_task(uint64_t id, uint32_t timeout, const destroy_func_t &cb) {
        task_sptr tp = make_shared<timer_task>(id, timeout, cb);
        tp->set_release_cb(bind(&time_wheel::remove_task, this, id));
        int pos = (_tick_idx + timeout) % _wheel_capacity;
        _wheel[pos].push_back(tp);
        _tasks.insert({id, task_wptr(tp)});
    }

    void remove_task(uint64_t id) {
        if(!has_timer(id)) {
            // 任务不存在
            return;
        }
        _tasks.erase(id);
    }
    
    void task_refresh(uint64_t id) {
        if(!has_timer(id)) {
            // 任务不存在
            return;
        }
        task_sptr task = _tasks[id].lock();
        if(task == nullptr) {
            INFO_LOG("void task_refresh(uint64_t id) err, id: %ld", id);
            return;
        }
        uint32_t new_pos = (_tick_idx + task->get_timeout()) % _wheel_capacity;
        _wheel[new_pos].push_back(task);
    }

    void task_cancel(uint64_t id) {
        if(!has_timer(id)) {
            // 任务不存在
            return;
        }
        task_sptr task = _tasks[id].lock();
        if(task == nullptr) {
            INFO_LOG("void task_cancel(uint64_t id) err, id: %ld", id);
            return;
        }
        task->cancel_task();
    }

    void tick() {
        _tick_idx = (_tick_idx + 1) % _wheel_capacity;
        for(const auto &task : _wheel[_tick_idx]) {
            if(!task->is_canceled()) {
                _tasks.erase(task->get_id());
            }
        }
        _wheel[_tick_idx].clear();
    }

    void on_time() {
        uint64_t times = read_timerfd();
        for(int i = 0; i < times; ++i) {
            tick();
        }
    }

public:
    time_wheel(event_loop *evl) 
        :_timerfd(get_timerfd()), _timer_ch(make_unique<channel>(evl, _timerfd)), _evl(evl),
        _tick_idx(0), _wheel_capacity(60), _wheel(_wheel_capacity) 
    { 
        _timer_ch->set_read_callback(bind(&time_wheel::on_time, this));
        _timer_ch->enable_read();
    }
    
    /*定时器中有个_timers成员，定时器信息的操作有可能在多线程中进行，因此需要考虑线程安全问题*/
    /*如果不想加锁，那就把对定期的所有操作，都放到一个线程中进行*/
    // 用到了event_loop中的接口，所以需要放到event_loop类后定义
    void add_task_inloop(uint64_t id, uint32_t timeout, const destroy_func_t &cb);

    void task_refresh_inloop(uint64_t id);

    void task_cancel_inloop(uint64_t id);

    // 存在线程安全，只能在模块内调用，evl线程中调用
    bool has_timer(uint64_t id) const {
        return _tasks.count(id) > 0;
    }
    
private:
    // timerfd，在系统中设置超时时间，每秒钟超时一次
    // 超时后回调执行事件处理函数，然后tick指针向后走一步
    int _timerfd;
    // 管理timerfd
    unique_ptr<channel> _timer_ch;
    // 与event_loop进行关联
    event_loop *_evl;
    // 秒针，走到哪里就执行哪里的销毁任务
    int _tick_idx;
    // 秒盘最大刻度
    int _wheel_capacity;
    // 时间轮 
    vector<vector<task_sptr>> _wheel;
    // 用来存放时间轮中的定时器对象信息，当执行销毁任务后需要把相应的对象从中移除
    // 注意这里必须有weak_ptr，否则它会影响时间轮中任务对象生命周期，进而导致引用计数减不到0
    // 无法执行销毁任务
    unordered_map<uint64_t, task_wptr> _tasks;
};

using functor = function<void()>;
class event_loop {
private:
    static int get_event_fd() {
        int fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        if (fd == -1) {
            ERROR_LOG("Failed to get event fd");
            exit(errno);
        }
        return fd;
    }

    void read_efd() {
        uint64_t ret = 0;
        ssize_t n = read(_event_fd, &ret, sizeof(ret));
        if (n != sizeof(ret)) {
            if(n == EINTR || n == EAGAIN) {
                return;
            }
            ERROR_LOG("Failed to read event fd");
            exit(errno);
        }
    }
    
    // 防止epoller因为没有事件就绪而阻塞
    void write_efd() {
        uint64_t val = 1;
        ssize_t n = write(_event_fd, &val, sizeof(val));
        if (n != sizeof(val)) {
            if(n == EINTR || n == EAGAIN) {
                return;
            }
            ERROR_LOG("Failed to read event fd");
            exit(errno);
        }
    }

    void run_tasks() {
        vector<functor> _funcs;
        {
            unique_lock<mutex> lock(_mtx);
            _tasks.swap(_funcs);
        }
        for (const auto &func : _funcs) {
            func();
        }
    }

public:
    event_loop()
        : _id(this_thread::get_id()), 
        _event_fd(get_event_fd()), 
        _efd_ch(make_unique<channel>(this, _event_fd)),
        _epoller(make_unique<epoller>()),
        _tw(make_unique<time_wheel>(this))
    {
        _efd_ch->set_read_callback(bind(&event_loop::read_efd, this));
        _efd_ch->enable_read();
    }
    
    // 主循环，监控、等待、处理
    void start() {
        while(true) {
            vector<channel *> readys = _epoller->poller();
            for (const auto& ch : readys) {
                ch->handle_event();
            }
            run_tasks();
        }
    }

    bool is_in_loop() const {
        return _id == this_thread::get_id();
    }

    void run_in_loop(const functor& cb) {
        if (is_in_loop()) {
            cb();
            return;
        } 
        add_to_tasks(cb);
        // 唤醒可能的epoll阻塞
        write_efd();
    }

    void add_to_tasks(const functor& cb) {
        {
            unique_lock<mutex> lock(_mtx);
            _tasks.push_back(cb);
        }
        write_efd();
    }

    void update(channel *ch) {
        _epoller->update(ch);
    }

    void remove(channel *ch) {
        _epoller->remove(ch);
    }

    // 增加定时器相关接口
    void add_timer(uint64_t id, uint32_t timeout, const destroy_func_t &cb) {
        _tw->add_task_inloop(id, timeout, cb);
    }

    void task_refresh(uint64_t id) {
        _tw->task_refresh_inloop(id);
    }

    void task_cancel(uint64_t id) {
        _tw->task_cancel_inloop(id);
    }

    bool has_timer(uint64_t id) const {
        return _tw->has_timer(id);
    }

private:
    thread::id _id;
    int _event_fd;
    unique_ptr<channel> _efd_ch;
    unique_ptr<epoller> _epoller;
    unique_ptr<time_wheel> _tw;
    mutex _mtx;
    vector<functor> _tasks;
};

// 一个eventloop对应一个线程
class loop_thread {
public:
    loop_thread()
        :_evl(nullptr), _thread(thread(&loop_thread::thread_entry, this))
    {}

    // 这个模块是将线程和loop整合在一块的
    // 那么当外界有一个新连接到来时，就需要给它分配一个loop
    // 所以需要提供一个获取loop对象的接口
    // 这样某个新连接就可以与某个loop关联起来了
    // 这样该连接再进行各种操作时都会在当前线程中执行
    event_loop *get_loop() {
        {
            unique_lock<mutex> lock(_mutex);
            // 为空就一直阻塞
            _cond.wait(lock, [&](){ return _evl != nullptr; });
        }
        return _evl;
    }

private:
    // 线程入口函数
    void thread_entry() {
        event_loop evl;
        {
            unique_lock<mutex> lock(mutex);
            _evl = &evl;
            _cond.notify_all();
        }
        evl.start();
    }

private: 
    // 互斥锁和条件变量用于实现获取loop的同步，
    // 避免线程创建了，但是loop对象还没有被初始化，此时获取loop的情况
    // 只有实例化完成之后才可以让其获取
    mutex _mutex;
    condition_variable _cond;
    // 必须设置为指针，需要在对应的线程内部去实例化，这样一个eventloop就绑定了一个线程
    // 而不是在当前线程中，否则绑定的就是主线程了
    event_loop *_evl;
    // eventloop对应的线程
    thread _thread;
};

class loop_thread_pool {
public:
    loop_thread_pool(event_loop *main_loop)
        :_thread_count(0), _loop_idx(0), _main_loop(main_loop) 
    {}

    void set_thread_count(int count) {
        _thread_count = count;
    }

    // 创建从属线程
    void create() {
        if(_thread_count > 0) {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for(int i = 0; i < _thread_count; ++i) {
                _threads[i] = new loop_thread;
                _loops[i] = _threads[i]->get_loop();
            }
        }
    }

    event_loop *next_loop() {
        if(_thread_count == 0) {
            return _main_loop;
        }
        _loop_idx = (_loop_idx + 1) % _thread_count;
        return _loops[_loop_idx];
    }
private:
    // 从属线程的数量
    int _thread_count;
    // 主loop，负责新连接
    // 若没有从属线程，则主线程同时负责新连接和对应连接的处理
    event_loop *_main_loop;
    // 保存所有对象
    vector<loop_thread*> _threads;
    // 分配loops中的event_loop线程
    vector<event_loop*> _loops;
    int _loop_idx;
};

class Any {
public:
    Any()
       : _val(nullptr)
    { }

    Any(const Any &other) {
        if (this != &other) {
            // 深拷贝
            _val = other._val == nullptr ? nullptr : other._val->clone();
        }
    }

    template<class T>
    Any(const T &val) 
        : _val(new derived<T>(val))
    {  }
 
    void swap(Any other) {
        std::swap(_val, other._val);
    }

    Any &operator=(const Any &other) {
        if (this != &other) {
            swap(other);
        }
        return *this;
    }
    
    template<class T>
    Any &operator=(const T &v) {
        swap(v);
        return *this;
    } 

    template<class T>
    T* get() const {
        // 先判断类型是否一致
        assert(typeid(T) == _val->type());
        derived<T> *d = dynamic_cast<derived<T> *>(_val);
        return d == nullptr ? nullptr : d->get();
    }

    ~Any() {
        delete _val;
    }

private:
    class base {
    public:
        virtual ~base() {
            // 保证所有子类都有析构函数
        }
        virtual const type_info& type() const = 0;
        virtual base* clone() const = 0;
    };

    template<class T>
    class derived : public base {
    public:
        derived(const T &val) 
            :_val(val)
        {}

        // 获取子类保存的数据类型
        virtual const type_info& type() const {
            return typeid(_val);
        }

        //  复制一个子类对象
        virtual base* clone() const {
            return new derived(*this);
        }
    
        T* get() {
            return &_val;
        }

    private:
        T _val;
    };

private:
    base *_val;
};

enum e {
    // 连接建立成功待处理状态
    CONNECTING,
    // 连接建立完成，各种设置完成可以进行通信的状态
    CONNECTED,
    // 待关闭状态，没有数据待处理就关闭
    DISCONNECTING,
    // 关闭状态
    DISCONNECTED
};
using con_stat = e;

class connection;
using con_ptr = shared_ptr<connection>;
class connection : public enable_shared_from_this<connection> { 
    using connected_callback = function<void(const con_ptr&)>;
    using closed_callback = function<void(const con_ptr&)>;
    using handle_messge_callback = function<void(const con_ptr&, buffer*)>;
    using any_event_callback = function<void(const con_ptr&)>;
public:
    connection(event_loop *evp, uint64_t con_id, int sock_fd)
        :_evp(evp), _status(CONNECTING), _con_id(con_id), _sock_fd(sock_fd), 
        _socket(sock_fd), _channel(evp, sock_fd), _enable_inactive_destroy(false)
    {
        _channel.set_read_callback(bind(&connection::handle_read, this));
        _channel.set_write_callback(bind(&connection::handle_write, this));
        _channel.set_close_callback(bind(&connection::handle_close, this));
        _channel.set_error_callback(bind(&connection::handle_error, this));
        _channel.set_event_callback(bind(&connection::handle_event, this));
        // 构造函数中不能启动读事件监控，因为在构造期间就有可能有事件就绪，若对象没有完全初始化则会导致错误
    }

    ~connection() {
        DEBUG_LOG("connection shutdown");
    }

    int get_fd() const {
        return _sock_fd;
    }

    uint64_t get_id() const {
        return _con_id;
    }

    bool is_connected() const {
        return _status == CONNECTED;
    }

    // 把数据放入到发送缓冲区并且启动读事件监控
    void send(const char *data, size_t len) {
        // 考虑data是临时空间，将任务压入任务池等到执行时可能这块空间已经销毁了
        // 所以需要把data持久化
        buffer buf;
        buf.writer(data, len);
        _evp->run_in_loop(bind(&connection::send_in_loop, this, buf));
    }

    // 提供给组件使用者的关闭接口
    // 并不是直接关闭，需要判断缓冲区状态最后关闭
    void shut_down() {
        _evp->run_in_loop(bind(&connection::shut_down_in_loop, this));
    }

    // 启动/取消非活跃超时销毁任务
    // 需要用户传递超时时间，秒为单位，然后添加销毁任务
    void enable_inactive_destroy(int second) {
        _evp->run_in_loop(bind(&connection::enable_inactive_destroy_in_loop, this, second));
    }

    void disable_inactive_destroy() {
        _evp->run_in_loop(bind(&connection::disable_inactive_destroy_in_loop, this));
    }

    // 协议切换，重置上下文和相应的回调处理函数
    // 这个接口必须要在绑定的loop中执行，防止新的事件触发后，处理时还是使用之前的协议进行处理
    void protocol_switching(const Any& context, const connected_callback& con_cb, const closed_callback& closed_cb, 
                            const handle_messge_callback& msg_cb, const any_event_callback& event_cb) {
        if(!_evp->is_in_loop()) {
            ERROR_LOG("Error is not in loop");
            exit(-1);
        }
        _evp->run_in_loop(bind(&connection::protocol_switching_in_loop, this, context, con_cb, closed_cb, msg_cb, event_cb));
    }
    
    // 连接建立就绪后，进行channel回调设置，启动读监控，调用connected_callback
    void established() {
        _evp->run_in_loop(bind(&connection::established_in_loop, this));
    }

    // 设置/获取上下文
    // 连接建立完成时设置 
    void set_context(const Any &context) {
        _context = context;
    }

    // 获取上下文
    Any &get_context() {
        return _context;
    }

    // 设置回调，由用户设置
    void set_connected_callback(const connected_callback& con_cb) {
        _con_cb = con_cb;
    }

    void set_closed_callback(const closed_callback& closed_cb) {
        _closed_cb = closed_cb;
    }

    void set_handle_messge_callback(const handle_messge_callback& msg_cb) {
        _msg_cb = msg_cb;
    }

    void set_any_event_callback(const any_event_callback& event_cb) {
        _any_event_cb = event_cb;
    }

    void set_server_closed_callback(const closed_callback& server_closed_cb) {
        _server_closed_cb = server_closed_cb;
    }

private:
    // 上述接口的实际执行需要保证是在当前所绑定的线程中进行的
    // 这个并不是实际的发送数据接口，而只是把数据放入到发送缓冲区中，启动写事件监控
    // 实际发送操作是在写事件触发后再进行的
    void send_in_loop(const buffer buf) {
        // 这种状态什么也不做
        if(_status == DISCONNECTED) {
            return;
        }
        // 写入，启动读事件监控
        _outbuf.buffer_writer(buf);
        if(!_channel.writeable()) {
            _channel.enable_write();
        }
    }

    // 在实际关闭之前需要判断缓冲区中是否还有数据待处理，最后关闭
    void shut_down_in_loop() {
        if(_inbuf.get_readable_size() > 0) {
            if(_msg_cb) {
                _msg_cb(shared_from_this(), &_inbuf);
            }
        }
        if(_outbuf.get_readable_size() > 0) {
            if(!_channel.writeable()) {
                _channel.enable_write();
            }
        }
        destroy_in_loop();
    }

    // 实际的释放接口
    void destroy_in_loop() {
        // 修改连接状态为disconnected
        _status = DISCONNECTED;
        // 关闭channel和socket
        _channel.remove();
        _socket.close();
        // 若有定时任务则取消，避免引用野指针
        _evp->task_cancel(_con_id);
        // 调用用户设置的关闭回调函数
        if(_closed_cb) {
            _closed_cb(shared_from_this());
        }
        // 再调用服务器的关闭函数清理连接信息
        if(_server_closed_cb) {
            _server_closed_cb(shared_from_this());
        }
    }

    // 启动非活跃连接销毁释放
    void enable_inactive_destroy_in_loop(int second) {
        _enable_inactive_destroy = true;
        // 如果存在就刷新，否则添加
        if(_evp->has_timer(_con_id)) {
            _evp->task_refresh(_con_id);
        }
        else {
            _evp->add_timer(_con_id, second, bind(&connection::destroy_in_loop, this));
        }
    }

    void disable_inactive_destroy_in_loop() {
        _enable_inactive_destroy = false;
        // 取消定时任务
        _evp->task_cancel(_con_id);
    }

    // 协议切换
    void protocol_switching_in_loop(const Any& context, const connected_callback& con_cb, const closed_callback& closed_cb, 
                                    const handle_messge_callback& msg_cb, const any_event_callback& event_cb) {
        // 改变了上下文和相应的处理函数就相当于完成了切换
        _context = context;
        _con_cb = con_cb;
        _closed_cb = closed_cb;
        _msg_cb = msg_cb;
        _any_event_cb = event_cb;
    }

    void established_in_loop() {
        // 修改连接状态，设置为connected
        if(_status != CONNECTING) {
            ERROR_LOG("status err");
            exit(-1);
        }
        _status = CONNECTED;
        // 启动读事件监控
        _channel.enable_read();
        // 调用用户设置的回调函数
        if(_con_cb) {
            _con_cb(shared_from_this());
        }
    }

    // 五个channel内的回调函数设置
    // 至于活跃度的刷新需要设置在事件处理完毕后进行
    // 因为如果先刷新活跃的，而事件执行的时间超过了超时时间会去执行销毁任务
    // 这样不符合预期，所以需要后刷新活跃度

    // 描述符可读事件触发后调用的函数，接收sk缓冲区的数据到接收缓冲区中，然后调用msg_cb进行业务处理
    void handle_read() {
        char buff[65536];
        ssize_t n = _socket.recv(buff, sizeof(buff) - 1, false);
        if(n < 0) {
            // 出错了不能直接关闭，需要检测是否还有数据待发送
            // 状态设置为待关闭
            shut_down_in_loop();
            return;
        }
        // 放入接收缓冲区
        _inbuf.writer(buff, n);
        // 有数据就需要进行处理
        if(_inbuf.get_readable_size() > 0) {
            // shared_from_this函数是 C++ 标准库中 std::enable_shared_from_this 类模板的成员函数
            // 它用于从当前对象（即派生自 std::enable_shared_from_this 的类的对象）生成一个 std::shared_ptr 智能指针
            // 该智能指针与当前对象共享所有权
            // 然而，shared_from_this 的使用有一个关键的前提条件：当前对象必须已经被一个 std::shared_ptr 所管理
            // 若要使用需要继承enable_shared_from_this<当前类型>
            _msg_cb(shared_from_this(), &_inbuf);
        }
    }

    // 描述符可写事件触发后调用的函数，将发送缓冲区中的数据进行发送
    void handle_write() {
        ssize_t n = _socket.send(_outbuf.get_read_pos(), _outbuf.get_readable_size(), false);
        if(n < 0) {
            // 出错了若接收缓冲区还有数据则进行处理
            if(_inbuf.get_readable_size() > 0) {
                _msg_cb(shared_from_this(), &_inbuf);
            }
            // 最后实际关闭连接，不能再绕回到shut_down_in_loop的逻辑了
            destroy_in_loop();
            return;
        }
        // 移动到下一次读位置
        _outbuf.move_read_pos(_outbuf.get_readable_size());
        // 由于写事件经常是就绪的，这会导致epoll经常返回但有没有实际的数据可写，影响效率
        // 所以当没有数据待发送时需要关闭写事件的关心
        if(_outbuf.get_readable_size() == 0) {
            _channel.disable_write();
            // 如果还是待关闭状态，则直接关闭连接
            if(_status == DISCONNECTING) {
                destroy_in_loop();
            }
        }
    }

    // 挂断事件触发后调用的函数
    void handle_close() {
        //若接收缓冲区还有数据则进行处理
        if(_inbuf.get_readable_size() > 0) {
            _msg_cb(shared_from_this(), &_inbuf);
        }
        // 最后实际关闭连接
        destroy_in_loop();
    }

    // 错误事件触发后调用的函数
    void handle_error() {
        handle_close();
    }

    // 任意事件触发后调用的函数
    void handle_event() {
        // 1. 若启动了非活跃超时销毁任务则需要刷新活跃度
        if(_enable_inactive_destroy) {
            _evp->task_refresh(_con_id);
        }
        // 2. 调用组件使用者设置的任意事件回调
        if(_any_event_cb) {
            _any_event_cb(shared_from_this());
        }
    }
private:
    // 为了保证操作是线程安全的，需要把连接对象绑定到一个eventloop上
    // 而loop则对应的一个线程
    event_loop *_evp;
    // 连接状态
    // 若要关闭一个连接应该直接关闭吗？
    // 显然不是，只有当缓冲区中没有数据待处理和待发送时才应该实际关闭连接
    // 所以需要一种状态信息来标记这种情况
    // 除此之外还有其它状态等
    con_stat _status;

    // 每个连接对象的唯一id，便于管理和查找
    // 同时也可以作为定时器任务id，只要具有唯一性即可
    // 当有事件触发时需要id来刷线对应任务的活跃度
    uint64_t _con_id;

    // 套接字的管理，要有套接字文件描述符和对应的操作
    int _sock_fd;
    Socket _socket;
    // 连接事件的管理
    channel _channel;

    // 从socket的缓冲区中接收数据
    // 可能当前数据包接收不完整，所以是需要在用户层维护一个接收缓冲区的
    // 依次拼接再用户缓冲区的后面，足够一个包时再进行处理
    // 亦或者若sk的输入缓冲区满了，数据无法写入，只能阻塞
    // 所以需要及时把数据取出来放入用户缓冲区中
    buffer _inbuf;
    // 向socket的发送缓冲区中写入数据，等待发送
    // 若sk的发送缓冲区数据满了再写会导致阻塞，所以需要在用户层维护一个发送缓冲区
    // 等待可写事件就绪再进行数据写入 
    buffer _outbuf;

    // 上下文管理
    // 由于每个协议的数据请求格式都不一样
    // 所以是需要一种通用结构来记录不同协议的数据以及数据处理的上下文
    // 这样就方便后续有数据到来时能继续进行处理
    Any _context;

    // 回调函数的设置
    // 连接建立成功之后、关闭连接之前、接收到数据后以及任意事件触发后如何处理
    // 这都是由组件使用者来决定的，因此必须要提供接口来设置对应的回调函数
    connected_callback _con_cb;
    closed_callback _closed_cb;
    handle_messge_callback _msg_cb;
    any_event_callback _any_event_cb;
    // 这个关闭接口是服务器组件内部设置的，因为服务器是所有连接的管理者
    // 一旦某个连接关闭就应该从管理的地方彻底移除
    closed_callback _server_closed_cb;

    // 一个用于判断是否开启非活跃连接销毁的标志位
    // 默认false
    bool _enable_inactive_destroy;
};

class acceptor {
    using acceptor_callback = function<void(int)>;
public:
    // 必须要在设置了事件回调函数之后启动读事件监控
    // 否则可能会造成事件就绪后由于没设置回调而导致资源泄漏
    acceptor(event_loop *evl, uint16_t port = default_port)
        :_evl(evl), _lis_sock(create_server(port)), _channel(evl, _lis_sock.get_sockfd())
    {
        _channel.set_read_callback(std::bind(&acceptor::handle_read, this));
    }

    void listen() {
        _channel.enable_read();
    }

    void set_acceptor_callback(const acceptor_callback &acceptor_cb) {
        _accept_cb = acceptor_cb;
    }
private:
    int create_server(uint16_t port = default_port) {
        _lis_sock.create_server(port);
        return _lis_sock.get_sockfd();
    }

    void handle_read() {
        int newfd = _lis_sock.accept();
        if(newfd == -1) {
            return;
        }
        if(_accept_cb) {
            _accept_cb(newfd);
        }
    }
private:
    // 创建监听套接字
    Socket _lis_sock;
    // 对监听套接字进行事件监控
    event_loop *_evl;
    // 对监听套接字进行事件管理
    channel _channel;
    acceptor_callback _accept_cb;
};

class tcp_server {
    using connected_callback = function<void(const con_ptr&)>;
    using closed_callback = function<void(const con_ptr&)>;
    using handle_messge_callback = function<void(const con_ptr&, buffer*)>;
    using any_event_callback = function<void(const con_ptr&)>;
public:
    tcp_server(uint16_t port):
        _port(port),
        _next_id(0),
        _enable_inactive_destroy(false),
        _acceptor(&_main_loop, port),
        _thread_pool(&_main_loop)
    {
        _acceptor.set_acceptor_callback(std::bind(&tcp_server::new_connection, this, placeholders::_1));
    }

    void set_thread_count(int count) {
        _thread_pool.set_thread_count(count);
    }

    void set_connected_callback(const connected_callback& con_cb) {
        _con_cb = con_cb;
    }

    void set_closed_callback(const closed_callback& closed_cb) {
        _closed_cb = closed_cb;
    }

    void set_handle_messge_callback(const handle_messge_callback& msg_cb) {
        _msg_cb = msg_cb;
    }

    void set_any_event_callback(const any_event_callback& event_cb) {
        _any_event_cb = event_cb;
    }

    void enable_inactive_destroy(int timeout) {
        _timeout = timeout;
        _enable_inactive_destroy = true;
    }

    // 添加定时任务
    void run_after(const functor &task, int delay) {
        _main_loop.run_in_loop(bind(&tcp_server::run_after_in_loop, this, task, delay));
    }

    void start() {
        _thread_pool.create();  // 创建从属线程
        _acceptor.listen(); // 把监听套接字挂接到主线程中开始监控
        _main_loop.start();
    }
private:
    // 为新连接构造一个连接对象进行管理
    void new_connection(int newfd) {
        con_ptr con = make_shared<connection>(_thread_pool.next_loop(), _next_id, newfd);
        con->set_connected_callback(_con_cb);
        con->set_handle_messge_callback(_msg_cb);
        con->set_closed_callback(_closed_cb);
        con->set_any_event_callback(_any_event_cb);
        con->set_server_closed_callback(bind(&tcp_server::remove_connection, this, placeholders::_1));

        if(_enable_inactive_destroy) {
            con->enable_inactive_destroy(5); 
        }
        con->established(); // 就绪初始化
        _conns.insert({_next_id, con});
        ++_next_id;
    }

    // 移除连接信息
    void remove_connection(const con_ptr &con) {
        _main_loop.run_in_loop(bind(&tcp_server::remove_connection_in_loop, this, con));
    }

    void remove_connection_in_loop(const con_ptr &con) {
        _conns.erase(con->get_id());
    }   

    void run_after_in_loop(const functor &task, int delay) {
        _next_id++;
        _main_loop.add_timer(_next_id, delay, task);
    }
private:
    uint16_t _port; // 服务器要监听的端口
    uint64_t _next_id;   // 连接和定时任务id
    int _timeout;       // 连接超时时间
    bool _enable_inactive_destroy;  // 一个用于判断是否开启非活跃连接销毁的标志位, 默认false
    event_loop _main_loop; // 主线程，负责监听事件的处理
    acceptor _acceptor; // 用来管理监听套接字
    loop_thread_pool _thread_pool;  // 从属线程池
    unordered_map<uint64_t, con_ptr> _conns; // 管理所有的连接对象

    // 对应的回调函数，用户设置给server模块，再由server模块设置给连接模块
    connected_callback _con_cb;
    closed_callback _closed_cb;
    handle_messge_callback _msg_cb;
    any_event_callback _any_event_cb;
};

// 由于先编译的channel模块，编译器不知道eventloop中的epoller中相关函数的实现
// 所以需要先定义eventloop中的epoller中的函数，后面再定义channel中的 
//移除监控
void channel::remove() {
    _ev_loop->remove(this);
}

void channel::update() {
    _ev_loop->update(this);  // 告诉epoll来更新channel
}

void time_wheel::add_task_inloop(uint64_t id, uint32_t timeout, const destroy_func_t &cb) {
        _evl->run_in_loop(bind(&time_wheel::add_task, this, id, timeout, cb));
}

void time_wheel::task_refresh_inloop(uint64_t id) {
    _evl->run_in_loop(bind(&time_wheel::task_refresh, this, id));
}

void time_wheel::task_cancel_inloop(uint64_t id) {
    _evl->run_in_loop(bind(&time_wheel::task_cancel, this, id));
} 

class net_work {
public:
    net_work() {
        INFO_LOG("net_work");
        // 当连接断开之后，如果另一方此时还进行send发生数据时系统会发生一个sigpipe信号来表示异常
        // 默认情况下该信号会导致程序的退出，这时可以对其进行忽略，不让程序退出
        // 这样当客户端断开连接时，如果服务器还给其send时服务器就不会退出了
        signal(SIGPIPE, SIG_IGN);
    }
};
static net_work nw;